package cn.chengpengper.rpc.provider.common.manager;

import cn.chengpengper.rpc.constants.RpcConstants;
import cn.chengpengper.rpc.protocol.RpcProtocol;
import cn.chengpengper.rpc.protocol.enums.RpcType;
import cn.chengpengper.rpc.protocol.header.RpcHeader;
import cn.chengpengper.rpc.protocol.header.RpcHeaderFactory;
import cn.chengpengper.rpc.protocol.response.RpcResponse;
import cn.chengpengper.rpc.provider.common.cache.ProviderChannelCache;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;

import java.util.Set;

/**
 * @author chengpeng.hu
 */
@Slf4j
public class ProviderConnectionManager {
    /**
     * 扫描并移除不活跃的连接
     */
    public static void scanNotActiveChannel() {
        Set<Channel> channelCache = ProviderChannelCache.getChannelCache();
        if (channelCache == null || channelCache.isEmpty()) {
            return;
        }
        channelCache.forEach((channel) -> {
            if (!channel.isOpen() || !channel.isActive()) {
                channel.close();
                ProviderChannelCache.remove(channel);
            }
        });
    }

    /**
     * 发送ping消息
     */
    public static void broadcastPingMessageFromProvider() {
        Set<Channel> channelCache = ProviderChannelCache.getChannelCache();
        if (channelCache == null || channelCache.isEmpty()) {
            return;
        }
        RpcHeader header = RpcHeaderFactory.getRequestHeader(RpcConstants.SERIALIZATION_PROTOSTUFF, RpcType.HEARTBEAT_FROM_PROVIDER);
        RpcProtocol<RpcResponse> responseRpcProtocol = new RpcProtocol<>();
        RpcResponse rpcResponse = new RpcResponse();
        rpcResponse.setResult(RpcConstants.HEARTBEAT_PING);
        responseRpcProtocol.setHeader(header);
        responseRpcProtocol.setBody(rpcResponse);
        channelCache.forEach((channel) -> {
            if (channel.isOpen() && channel.isActive()) {
                //1、检查是否超过次数
                boolean overflow = ProviderChannelCache.isWaitTimesOverflow(channel);
                if (overflow) {
                    // 关闭通道并清除缓存
                    closeChannelAndClear(channel);
                } else {
                    // 正常发送心跳
                    log.info("send heartbeat message to consumer, the consumer is: {}, the heartbeat message is: {}", channel.remoteAddress(), rpcResponse.getResult());
                    channel.writeAndFlush(responseRpcProtocol);
                    int count = ProviderChannelCache.increaseTimes(channel);
                    log.info("服务提供者给服务消费者【{}】发送心跳，当前心跳响应等待【{}】次", channel.remoteAddress(), count);
                }
            }
        });
    }

    private static void closeChannelAndClear(Channel channel) {
        log.info("服务消费者 {} 超过3次没有回复服务提供者心跳，将关闭通道并删除缓存", channel.remoteAddress());
        channel.close();
        ProviderChannelCache.remove(channel);
    }
}
